package examples

import (
	"fmt"
	"time"

	"gitee.com/yijimu/gin-frame/utils/kafka"
	"github.com/Shopify/sarama"
)

// 同步生产者例子
func main() {
	producerClient, err := kafka.NewSyncProducerClient([]string{"192.168.3.180:19092"}, func(config *sarama.Config) {
		config.Producer.Flush.Frequency = 500 * time.Millisecond  // 每0.5s冲洗一批
		config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
		config.Producer.Partitioner = sarama.NewRandomPartitioner // 设置生产者的数据写入哪个分区（此处随机设置一个分区）
		//多个分区保证消息顺序,可以采用下面的方法(注释掉上面一行)，这样就可以避免一个topic只分配一个分区带来的效率问题
		//config.Producer.Partitioner = sarama.NewHashPartitioner // 多个分区保证消息顺序可以设置选择分区的策略为Hash，在生成消息之前，设置消息的Key值，见下
	})

	if err != nil {
		fmt.Println(err.Error())
		return
	}

	defer func(producerClient sarama.SyncProducer) {
		_ = producerClient.Close()
	}(producerClient)

	p, o, e := producerClient.SendMessage(&sarama.ProducerMessage{
		Topic:     "huangzhilin",
		Value:     sarama.ByteEncoder("我爱你哦"),
		Timestamp: time.Now(),
		//对Key进行Hash，同样的Key每次都落到一个分区，这样消息是有序的
		//需要配合 config.Producer.Partitioner = sarama.NewHashPartitioner 如上
		Key: sarama.StringEncoder(12344), //Kafka客户端会根据Key进行Hash，我们通过把接收用户ID作为Key，这样就能让所有发给某个人的消息落到同一个分区了，也就有序了
	})
	if e != nil {
		fmt.Println(e.Error())
		return
	}
	fmt.Println(p, o)
}
